package midas.mw.crt.jxta;

import midas.mw.cfg.CfgError;
import midas.mw.cfg.CfgLogging;
import midas.mw.cfg.log.Logger;
import midas.mw.crt.Constants;
import midas.mw.crt.CrtException;
import midas.mw.crt.MD.PriorityTable;
import midas.mw.crt.common.IPMessage;
import midas.mw.crt.interfaces.CrtCallback;
import midas.mw.crt.presence.util.MD5ID;
import net.jxta.document.AdvertisementFactory;
import net.jxta.endpoint.Message;
import net.jxta.endpoint.MessageElement;
import net.jxta.id.ID;
import net.jxta.peergroup.PeerGroup;
import net.jxta.pipe.InputPipe;
import net.jxta.pipe.PipeMsgEvent;
import net.jxta.pipe.PipeMsgListener;
import net.jxta.pipe.PipeService;
import net.jxta.protocol.PipeAdvertisement;
import java.io.IOException;
import java.util.Date;
import java.util.Iterator;
import java.util.Vector;



public class MessagePipe implements PipeMsgListener 
{
	private PipeService pipeService;
	
	private InputPipe inputPipe = null;
	
	private JxtaModule jxta = null;
	
	public PipeAdvertisement advertisement = null;
	/**
	* Constructor for the PipeServer object
	* 
	*/
	public MessagePipe(JxtaModule jxta) 
	{
		CfgLogging.logEvent(CfgLogging.LOG_MCALL, Logger.COMPONENT_CRT, CfgLogging.SEV_DEBUG, "MessagePipe(JxtaModule)");
		
		this.jxta = jxta;
	}
	
	/**
	* Dumps the message content to stdout
	*
	* @param msg the message
	* @param verbose dumps message element content if true
	*/
	/*public static void printMessageStats(Message msg, boolean verbose) 
	{
		try {
			CountingOutputStream cnt;
			ElementIterator it = msg.getMessageElements();
			System.out.println("------------------Begin Message---------------------");
			WireFormatMessage serialed = WireFormatMessageFactory.toWire(
					msg, new MimeMediaType("application/x-jxta-msg"), null);
			System.out.println("Message Size :" + serialed.getByteLength());
			while (it.hasNext()) {
				MessageElement el = it.next();
				String eName = el.getElementName();
				cnt = new CountingOutputStream(new DevNullOutputStream());
				el.sendToStream(cnt);
				long size = cnt.getBytesWritten();
				System.out.println("Element " + eName + " : " + size);
				if (verbose) {
					System.out.println("[" + el + "]");
				}
			}
			System.out.println("-------------------End Message----------------------");
		} catch (Exception e) {
			e.printStackTrace();
		}
	}*/
	/**
	* Creates the input pipe with this as the message listener
	*/
	public InputPipe start(PeerGroup peerGroup) 
	{
		CfgLogging.logEvent(CfgLogging.LOG_MCALL, Logger.COMPONENT_CRT, CfgLogging.SEV_DEBUG, "MessagePipe.start(PeerGroup)");
		
		try {					
			advertisement = (PipeAdvertisement)
			AdvertisementFactory.newAdvertisement(PipeAdvertisement.getAdvertisementType());
			ID messagePipeID = (net.jxta.id.ID)MD5ID.createPipeID(peerGroup.getPeerGroupID(),"MessageService", "Node1");			
			advertisement.setPipeID(messagePipeID);
			advertisement.setType(PipeService.UnicastType);
			advertisement.setName("MessagePipe");
			try {
	        	peerGroup.getDiscoveryService().publish(advertisement);	        	
	        } catch(Exception e){
	            
	            throw new RuntimeException("Critical Error pubishing our own advertisement");
	        }
			// Create the InputPipe and register this for message arrival
			// notification call-back
			pipeService = peerGroup.getPipeService();
			inputPipe = pipeService.createInputPipe(advertisement, this);
		} catch (IOException io) {
			io.printStackTrace();
			return null;
		}
		if (inputPipe == null) {
			CfgError.errorReporting(CfgError.ERR_OPER, Logger.COMPONENT_CRT, CfgError.SEV_FATAL, 
					"MessagePipe.start(). Cannot open InputPipe ", null);			
			return null;
			//System.exit(-1);
		}
		CfgLogging.logEvent(CfgLogging.LOG_INFO, Logger.COMPONENT_CRT, CfgLogging.SEV_INFORMAL,
					"MessagePipe.start(). Waiting for msgs on input pipe");
		return inputPipe;
	}
	/**
	* Closes the output pipe and stops the platform
	*/
	public void stop() 
	{
		CfgLogging.logEvent(CfgLogging.LOG_MCALL, Logger.COMPONENT_CRT, CfgLogging.SEV_DEBUG, "MessagePipe.stop()");
		
	// Close the input pipe
		inputPipe.close();
		//Stop JXTA
	}
	/**
	* PipeMsgListener interface for asynchronous message arrival notification
	*
	* @param event the message event
	*/
	public void pipeMsgEvent(PipeMsgEvent event) 
	{	
		CfgLogging.logEvent(CfgLogging.LOG_MCALL, Logger.COMPONENT_CRT, CfgLogging.SEV_DEBUG, "MessagePipe.pipeMsgEvent(PipeMsgEvent)");
		
		Message msg;
		try {
		// Obtain the message from the event
			msg = event.getMessage();
			if (msg == null) {
				CfgError.errorReporting(CfgError.ERR_OPER, Logger.COMPONENT_CRT, CfgError.SEV_WARNING, 
						"MessagePipe.pipeMsgEvent(). Received an empty message ", null);				
				return;
			}
			// dump the message content to screen
			//printMessageStats(msg, true);
		} catch (Exception e) {
			e.printStackTrace();
			return;
		}
		// get all the message elements
		Message.ElementIterator en = msg.getMessageElements();
		if (!en.hasNext()) {
			return;
		}
					
		MessageElement me1 = msg.getMessageElement(Constants.MIDAS, Constants.MESSAGE_TYPE);
		if (me1.toString().equals(Constants.IP_MESSAGE_QUESTION))
		{
			msg.removeMessageElement(me1);
			Iterator en2 = msg.getMessageElementsOfNamespace(Constants.MIDAS);
												
			String srcNode 	= ((MessageElement)en2.next()).toString();															
			
			CfgLogging.logEvent(CfgLogging.LOG_INFO, Logger.COMPONENT_CRT, CfgLogging.SEV_INFORMAL, "MessagePipe.pipeMsgEvent(PipeMsgEvent). " + 
					" \n\tType: " + me1.toString() +					
					" \n\tSource node: " + srcNode);
			
			IPMessage ipMessage = new IPMessage(jxta.getCrt().getOwnNodeId() ,srcNode, "", "", jxta.getCrt().getOwnNodeId().getBytes());		
			ipMessage.setMessageType(Constants.IP_MESSAGE_QUESTION);
			try {
				jxta.forwardMessage(ipMessage);
			} catch (CrtException crte){
				crte.printStackTrace();
			}
		}
		if (me1.toString().equals(Constants.IP_MESSAGE_PRESENCE))
		{
			msg.removeMessageElement(me1);
			Iterator en2 = msg.getMessageElementsOfNamespace(Constants.MIDAS);
												
			String srcNode 	= ((MessageElement)en2.next()).toString();															
			
			CfgLogging.logEvent(CfgLogging.LOG_INFO, Logger.COMPONENT_CRT, CfgLogging.SEV_INFORMAL, "MessagePipe.pipeMsgEvent(PipeMsgEvent). " + 
					" \n\tType: " + me1.toString() +					
					" \n\tSource node: " + srcNode);
			
			//jxta.getCrt().getNokiaPresenceThread().isPresent(srcNode);
			
		}
		else if (me1.toString().equals(Constants.IP_MESSAGE_BROADCAST))
		{
			msg.removeMessageElement(me1);
			Iterator en2 = msg.getMessageElementsOfNamespace(Constants.MIDAS);
											
			String srcNode 	= ((MessageElement)en2.next()).toString();												
			String srcApp 		= ((MessageElement)en2.next()).toString();			
			String dstApp 		= ((MessageElement)en2.next()).toString();																
			byte [] data		= ((MessageElement)en2.next()).getBytes(false);							
			
			CfgLogging.logEvent(CfgLogging.LOG_INFO, Logger.COMPONENT_CRT, CfgLogging.SEV_INFORMAL, "MessagePipe.pipeMsgEvent(PipeMsgEvent). " + 
					" \n\tType: " + me1.toString() +					
					" \n\tSource node: " + srcNode + 					
					" \n\tSource app: " + srcApp + 
					" \n\tDestination app : " + dstApp);								
					//" \n\tData: " + data);	
			
			//If it is a Central Node, forward the message
			if (jxta.getCentralNodeId() == null)
			{				
				try {
					IPMessage ipMessage = new IPMessage(srcNode ,null, srcApp, dstApp, data);					
					jxta.propagateMessage(ipMessage);					
										
					//Se lo manda tb. a los nodos JXME que tiene conectados
					ipMessage.setMessageType(Constants.IP_MESSAGE_BROADCAST);
					Vector jxmeNodes = jxta.getCrt().getNodeIdentifierSolver().getOwnJXMENodes();							
					for (int i=0; i<jxmeNodes.size(); i++)
					{						
						ipMessage.setDestinationNodeId((String)jxmeNodes.get(i));
						jxta.forwardMessage(ipMessage);
					}
					
				} catch (CrtException crte){
					crte.printStackTrace();
				}
			}	
			
			//Forward the message to the appropiate application/component			
			IPMessage ipMessage = new IPMessage(srcNode, null, srcApp, dstApp, data);
			ipMessage.setMessageType(Constants.IP_MESSAGE_BROADCAST);
			/*if (mID != null)
				ipMessage.setMessageId(mID);*/
			//First, we look for the destination among the registered components (but not the MW)
			CrtCallback crtUser = null;
			crtUser = (CrtCallback) jxta.getCrt().getCRTUser(dstApp);
			if (crtUser != null)
				crtUser.processIncomingMessage(ipMessage);	
			else
			{
				//We look for the destination between components
				crtUser = (CrtCallback) jxta.getCrt().getMiddlewareInterface().getRunningComponent(dstApp);
				if (crtUser != null)
					crtUser.processIncomingMessage(ipMessage);
				//We look for the destination between MIDAS applications
				else
				{				
					CfgError.errorReporting(CfgError.ERR_INFO, Logger.COMPONENT_CRT, CfgError.SEV_INFORMAL, "MessagePipe.pipeMsgEvent(PipeMsgEvent). " + 
							"The message destination user is not registered", null);
				}
			}
		} 		
		
		//Llega un MESSAGE_REQUEST
		else if (me1.toString().equals(Constants.IP_MESSAGE_REQUEST))
		{			
			msg.removeMessageElement(me1);
			Iterator en2 = msg.getMessageElementsOfNamespace(Constants.MIDAS);
			
			
			String srcNode 	= ((MessageElement)en2.next()).toString();			
			String dstNode 	= ((MessageElement)en2.next()).toString();							
			String srcApp 		= ((MessageElement)en2.next()).toString();			
			String dstApp 		= ((MessageElement)en2.next()).toString();					
			long time 			= Long.parseLong(((MessageElement)en2.next()).toString());			
			byte priority 		= Byte.parseByte(((MessageElement)en2.next()).toString());			
			int confirmation = Integer.parseInt(((MessageElement)en2.next()).toString());			
			Long mID    		= new Long(((MessageElement)en2.next()).toString());			
			int totalLength	= Integer.parseInt(((MessageElement)en2.next()).toString());			
			int numMessages	= Integer.parseInt(((MessageElement)en2.next()).toString());			
			int numOrder	= Integer.parseInt(((MessageElement)en2.next()).toString());			
			byte [] data	= ((MessageElement)en2.next()).getBytes(false);					
			
			CfgLogging.logEvent(CfgLogging.LOG_INFO, Logger.COMPONENT_CRT, CfgLogging.SEV_INFORMAL, "MessagePipe.pipeMsgEvent(PipeMsgEvent). " + 
					" \n\tType: " + me1.toString() +					
					" \n\tSource node: " + srcNode + 
					" \n\tDestination node: " + dstNode +
					" \n\tSource app: " + srcApp + 
					" \n\tDestination app : " + dstApp +
					" \n\tTime : " + time +
					" \n\tPriority : " + priority +
					" \n\tConfirmation : " + confirmation +
					" \n\tTotal length: " + totalLength +
					" \n\tNum messages : " + numMessages + 
					" \n\tNum order : " + numOrder +
					" \n\tMessage Id : " + mID);					
					//" \n\tData: " + data);	
			
			//Si es para otro nodo
			//	- Se modifica el messageId
			//	- Se incluye el campo FinalDestinationNodeId
			//	- Se encola para mandarlo
			IPMessage ipMessage = null;
			if (!dstNode.equals(jxta.getCrt().getOwnNodeId()))
			{				
				ipMessage = new IPMessage(srcNode, dstNode, srcApp, dstApp, data);
				ipMessage.setTotalLength(totalLength);
				ipMessage.setNumMessages(numMessages);
				ipMessage.setNumOrder(numOrder);
				if (confirmation == 1)
					ipMessage.setConfirmation(true);			
				//long myMessageId = Long.valueOf(mID).longValue() + 1000;
				//ipMessage.setMessageId(new Long(myMessageId));
				ipMessage.setMessageId(mID);
				ipMessage.setFinalDestinationNodeId(dstNode);				
				jxta.getCrt().getMessageDispatcher().sendMessage(ipMessage);
			}
			//Si es para este nodo
			//	- Se responde si es necesario
			//	- Se notifica a la aplicacion
			else
			{				
				if (confirmation == 1)
				{					
					ipMessage = new IPMessage(dstNode ,srcNode, dstApp, srcApp, "OK".getBytes());
					ipMessage.setMessageId(mID);
					ipMessage.setMessageType(Constants.IP_MESSAGE_RESPONSE);
					try {
						jxta.forwardMessage(ipMessage);
					} catch (CrtException crte){
						crte.printStackTrace();
					}
					ipMessage = null;
				}
				
				//Puede que sea un mensaje dividido en partes por lo que hay que ir almacenandolas
				if (totalLength > data.length)
				{					
					synchronized(srcNode)
					{
						ipMessage = jxta.getCrt().getMessageReceiver().getMessage(srcNode, mID);
						if (ipMessage == null)
						{											
							ipMessage = new IPMessage(srcNode, dstNode, srcApp, dstApp, new byte[totalLength]);					
							
							ipMessage.setMessageId(mID);
							ipMessage.setTotalLength(totalLength);
							ipMessage.setNumOrder(numOrder);
							System.arraycopy(data, 0, ipMessage.getMessageContent(), 0, data.length);			
							jxta.getCrt().getMessageReceiver().addMessage(ipMessage);
													
						}
						else
							System.arraycopy(data, 0, ipMessage.getMessageContent(), Constants.MAX_MESS_SIZE*(numOrder-1), data.length);						
					
						/*ipMessage = new IPMessage(srcNode, dstNode, srcApp, dstApp, data);
						ipMessage.setMessageId(new Long(mID));
						ipMessage.setTotalLength(totalLength);
						ipMessage.setNumOrder(numOrder);
						crt.getMessageReceiver().addMessagePart(ipMessage);*/
						if (numOrder == numMessages)					
						{
							//Es el ultimo												
							ipMessage = (IPMessage) jxta.getCrt().getMessageReceiver().getMessage(srcNode, ipMessage.getMessageId());					
							jxta.getCrt().getMessageReceiver().removeMessage(srcNode, ipMessage.getMessageId());
							//crt.getMessageReceiver().printTable();
						}
						else						
							return;
					}
				}				
				//Forward the message to the appropiate application/component				
				if (ipMessage == null)
				{
					ipMessage = new IPMessage(srcNode, dstNode, srcApp, dstApp, data);
					ipMessage.setTotalLength(totalLength);
					ipMessage.setMessageId(mID);					
				}
				
				if (ipMessage.getMessageContent().length != ipMessage.getTotalLength())
				{
					//System.out.println("NO TENGO EL MENSAJE COMPLETO");
				}
				else
				{
					//Forward the message to the appropiate application/component											
					//First, we look for the destination among the registered components (but not the MW)
					CrtCallback crtUser = null;
					crtUser = (CrtCallback) jxta.getCrt().getCRTUser(dstApp);
					if (crtUser != null)
						crtUser.processIncomingMessage(ipMessage);	
					else
					{
						//We look for the destination between components
						crtUser = (CrtCallback) jxta.getCrt().getMiddlewareInterface().getRunningComponent(dstApp);
						if (crtUser != null)
							crtUser.processIncomingMessage(ipMessage);
						//We look for the destination between MIDAS applications
					
						else
							CfgError.errorReporting(CfgError.ERR_INFO, Logger.COMPONENT_CRT, CfgError.SEV_INFORMAL, "MessagePipe.pipeMsgEvent(PipeMsgEvent). " + 
									"The message destination user is not registered", null);					
					}
				}		
			}
		} 		
		
		else if (me1.toString().equals(Constants.IP_MESSAGE_RESPONSE))
		{
			msg.removeMessageElement(me1);
			Iterator en2 = msg.getMessageElementsOfNamespace(Constants.MIDAS);
												
			String srcNode 	= ((MessageElement)en2.next()).toString();			
			String dstNode 	= ((MessageElement)en2.next()).toString();							
			String srcApp 		= ((MessageElement)en2.next()).toString();			
			String dstApp 		= ((MessageElement)en2.next()).toString();										
			Long mID    		= new Long(((MessageElement)en2.next()).toString());									
			byte [] data		= ((MessageElement)en2.next()).getBytes(false);							
			
			CfgLogging.logEvent(CfgLogging.LOG_INFO, Logger.COMPONENT_CRT, CfgLogging.SEV_INFORMAL, "MessagePipe.pipeMsgEvent(PipeMsgEvent). " + 
					" \n\tType: " + me1.toString() +					
					" \n\tSource node: " + srcNode + 
					" \n\tDestination node: " + dstNode +
					" \n\tSource app: " + srcApp + 
					" \n\tDestination app : " + dstApp +					
					" \n\tMessage Id : " + mID);					
					//" \n\tData: " + data);	
			
			if (!dstNode.equals(jxta.getCrt().getOwnNodeId()))
			{			
				IPMessage ipMessage = new IPMessage(srcNode, dstNode, srcApp, dstApp, data);
				//if (confirmation.equals("1"))
				//	ipMessage.setConfirmation(true);
				/*long realMessageId = 0;
				//Disminuimos en 2000 porque antes lo habiamos aumentado
				if (jxta.getCrt().getNodeIdentifierSolver().isJXMENode(dstNode))
					realMessageId = Long.valueOf(mID).longValue() - 1000;
				else
					realMessageId = Long.valueOf(mID).longValue() - 2000;
				ipMessage.setMessageId(new Long(realMessageId));*/
				ipMessage.setMessageId(mID);
				ipMessage.setMessageType(Constants.IP_MESSAGE_RESPONSE);
				jxta.getCrt().getMessageDispatcher().sendMessage(ipMessage);
			}
			else
			{				
				//Le borramos de la lista
				/*IPMessage receivedMessage = (IPMessage)jxta.getCrt().getMessageDispatcher().getMessageList().get(mID);
				if (receivedMessage != null)
				{
					jxta.getCrt().getMessageDispatcher().getTimeQueue().deleteMessage(receivedMessage);
					jxta.getCrt().getMessageDispatcher().getNodeList().deleteMessage(receivedMessage);
					jxta.getCrt().getMessageDispatcher().getMessageList().remove(mID);
				}
				
				jxta.getCrt().getMessageDispatcher().getNodeList().setStatus(srcNode, false);
				*/
				//Forward the message to the appropiate application/component			
				IPMessage ipMessage = new IPMessage(srcNode, dstNode, srcApp, dstApp, data);
				ipMessage.setMessageId(mID);
				//First, we look for the destination among the registered components (but not the MW)
				CrtCallback crtUser = null;
				crtUser = (CrtCallback) jxta.getCrt().getCRTUser(dstApp);
				if (crtUser != null)
					crtUser.processDeliveredMessage(ipMessage);	
				else
				{
					//We look for the destination between components
					crtUser = (CrtCallback) jxta.getCrt().getMiddlewareInterface().getRunningComponent(dstApp);
					if (crtUser != null)
						crtUser.processDeliveredMessage(ipMessage);
					//We look for the destination between MIDAS applications
					
					else
						CfgError.errorReporting(CfgError.ERR_INFO, Logger.COMPONENT_CRT, CfgError.SEV_INFORMAL, "MessagePipe.pipeMsgEvent(PipeMsgEvent). " + 
								"The message destination user is not registered", null);				
				}
				try {
					Object obj = jxta.getCrt().getMessageDispatcher().getNodeList().getNodeQueue(srcNode);
					if (obj instanceof PriorityTable)
						((PriorityTable)obj).getQueueThread().responseGot();		
				} catch (Exception e){e.printStackTrace();}
			}						
		} 		
		// get the message element in the name space PipeClient.MESSAGE_NAME_SPACE
		//MessageElement msgElement = msg.getMessageElement(MESSAGE_NAME_SPACE);
		// Get message
		if (me1.toString() == null) {
			CfgError.errorReporting(CfgError.ERR_OPER, Logger.COMPONENT_CRT, CfgError.SEV_WARNING, 
					"MessagePipe.pipeMsgEvent(PipeMsgEvent). Null msg received ", null);			
		} 
		else {
			Date date = new Date(System.currentTimeMillis());
			CfgLogging.logEvent(CfgLogging.LOG_INFO, Logger.COMPONENT_CRT, CfgError.SEV_INFORMAL,
					"MessagePipe.pipeMsgEvent(PipeMsgEvent) Message received at :" + date.toString());
			CfgLogging.logEvent(CfgLogging.LOG_INFO, Logger.COMPONENT_CRT, CfgError.SEV_INFORMAL,
					"MessagePipe.pipeMsgEvent(PipeMsgEvent) Message created at :" + me1.toString());
		}		
	}
}
